package com.aotain.project.apollo.bolt;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.math3.stat.regression.OLSMultipleLinearRegression;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;

public class LineRegressionBolt extends AbstractBolt{

	private Map<String,String> recordsMap = new HashMap<String,String>();
	private Map<String,ArrayList<String>> factorMap = new HashMap<String,ArrayList<String>>();//各目标Ip 回归因子
	private Map<String,double[]> kiMap = new HashMap<String,double[]>();//系数ki
	private Map<String,Double> standardErrorMap = new HashMap<String,Double>();
	private String dip;
	private Integer upstreamoctets;
	private Integer upstreampacket;
	private Integer pv;
	private OutputCollector _collector;
	private String regressorStr;
	private OLSMultipleLinearRegression regression;
	private double[] ki;
	private double preditValue;
	private String dateStr;
	private String rowKey ;
	HBaseRecordAdd hbaseInstance;



	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		try {

			if(isTickTuple(input)) {//满足时间条件
				emitCountingData();  //计算，入库
				_collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				_collector.ack(input);
			}
		} catch (Exception e) {
			Logger.getRootLogger().error(e);
			throw new FailedException("LineRegressionBolt Exception",e);
		}	
	}

	private void countInLocal(Tuple tuple) {
		dip = tuple.getStringByField("dip");
		upstreamoctets = tuple.getIntegerByField("upstreamoctets");
		upstreampacket = tuple.getIntegerByField("upstreampacket");
		pv = tuple.getIntegerByField("pv");
		dateStr = tuple.getStringByField("dateStr");
		ArrayList<String> arrList ;

		/*
		 * List : 连接数，包大小，包个数
		 */
		if(factorMap.get(dip) == null) {
			arrList = new ArrayList<String>();
		} else {
			arrList = factorMap.get(dip);
		}
		regressorStr = String.format("%s,%s,%s", pv,upstreamoctets,upstreampacket);
		arrList.add(regressorStr);
		factorMap.put(dip, arrList);
//		Logger.getLogger(LineRegressionBolt.class).info("####LineRegressionBolt:regressorStr####" + dip + "," +regressorStr);

		rowKey = String.format("%s_%s", dip,dateStr);
		if(kiMap.get(dip) != null) {
			ki = kiMap.get(dip);
			preditValue = ki[0] + upstreamoctets*ki[1] + upstreampacket*ki[2];
//			Logger.getLogger(LineRegressionBolt.class).info("####LineRegressionBolt:ki####" + ki);
			hbaseInstance.Add("SDS_SESSION_PREDIT", rowKey, "cf", "TRUTHPV", String.valueOf(pv));
			hbaseInstance.Add("SDS_SESSION_PREDIT", rowKey, "cf", "UPSTREAMOCTETS", String.valueOf(upstreamoctets));
			hbaseInstance.Add("SDS_SESSION_PREDIT", rowKey, "cf", "UPSTREAMPACKET", String.valueOf(upstreampacket));
			hbaseInstance.Add("SDS_SESSION_PREDIT", rowKey, "cf", "PREDIT", String.valueOf(preditValue));
		}
	}

	private void emitCountingData() {

		for(Entry<String,ArrayList<String>> entry : factorMap.entrySet()) {
			dip = entry.getKey();
			ArrayList<String> list = entry.getValue();
			double[] yList = new double[list.size()];
			double[][] xList = new double[list.size()][];
			for(int i=0 ;i<list.size();i++) {
				regressorStr = list.get(i);
				String[] splits = regressorStr.split(",");
				yList[i] = Double.parseDouble(splits[0]);
				xList[i] = new double[]{Double.parseDouble(splits[1]),Double.parseDouble(splits[2])};
			}
			
			try {
				if(xList.length > xList[0].length && yList.length > 10) {//样本记录数要大于 变量自变量个数
					regression.newSampleData(yList, xList);
					double[] ki = regression.estimateRegressionParameters();
					//				double standardError = regression.estimateRegressionStandardError();
					kiMap.put(dip, ki);
//					Logger.getLogger(LineRegressionBolt.class).info("####LineRegressionBolt:ki>#### " + dip + " "+ ki[0] + " " + ki[1] + " " + ki[2] + " length=" + ki.length);
				}
			}catch(Exception e) {
				
				System.out.println("####LineRegressionBolt:::Exception=>xList=" + Arrays.deepToString(xList) + "; yList=" + Arrays.toString(yList));
				Logger.getRootLogger().error(e);
//				throw new FailedException("LineRegressionBolt Exception",e);
			}
			//			standardErrorMap.put(dip, standardError);
			
		list.clear();//注意清空
		}

		factorMap.clear();
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub

	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,3600);
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		_collector = collector;
		regression = new OLSMultipleLinearRegression();
		hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);
	}

}
